
import bean.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author shall潇
 * @Date 2021/7/6
 * @Description		将温度按照高温、正常、低温分流
 *
 * 不采用 splitStream（旧）方式
 */
public class TransformTest4_MultipleStreamsSplit {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //使用侧输出流获得
        OutputTag<SensorReading> highTemp = new OutputTag<SensorReading>("highTemp"){};
        OutputTag<SensorReading> lowTemp = new OutputTag<SensorReading>("lowTemp"){};

        //DataStreamSource<String> dataStreamSource = env.socketTextStream("192.168.159.100", 7777);
        DataStream<String> dataStreamSource = env.readTextFile(TransformTest1_Base.FILE_PATH);
        SingleOutputStreamOperator<SensorReading> mapStream = dataStreamSource.map(line -> {
            String[] split = line.split(",");
            SensorReading sensorReading = new SensorReading(split[0], Long.valueOf(split[1]), Float.valueOf(split[2]));
            return sensorReading;
        });

        SingleOutputStreamOperator<SensorReading> process = mapStream.process(new MyProcessFunction2(highTemp,lowTemp));

        process.print("normal");

        // 通过getSideOutPut获取指定测输出流，参数：OutputTag
        DataStream<SensorReading> sideOutput1 = process.getSideOutput(highTemp);
        sideOutput1.print("high-temp");

        DataStream<SensorReading> sideOutput2 = process.getSideOutput(lowTemp);
        sideOutput2.print("low-temp");

        try {
            env.execute("process-2");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static class MyProcessFunction2 extends ProcessFunction<SensorReading, SensorReading> {
        private OutputTag<SensorReading> highTemp;
        private OutputTag<SensorReading> lowTemp;

        public MyProcessFunction2(OutputTag<SensorReading> highTemp, OutputTag<SensorReading> lowTemp) {
            this.highTemp = highTemp;
            this.lowTemp = lowTemp;
        }

        @Override
        public void processElement(SensorReading sensorReading, Context context, Collector<SensorReading> collector) throws Exception {
            /*高温、低温---选择放入测输出流*/
            if(sensorReading.getTemperature()>37){
                context.output(highTemp,sensorReading);
            }else if(sensorReading.getTemperature()<36){
                context.output(lowTemp,sensorReading);
            }else {
                /*正常温度---选择主流*/
                collector.collect(sensorReading);
            }
        }
    }
}
